/**
 * Copyright 2007-2008. Chongqing First Information & Network Co., Ltd. All
 * rights reserved. <a>http://www.cqfirst.com.cn</a>
 */
package com.cqcis.uip.base.framework.handlers.common;

import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.CountDownLatch;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.mina.common.ByteBuffer;
import org.apache.mina.common.IdleStatus;
import org.apache.mina.common.IoSession;
import org.apache.mina.common.WriteFuture;
import org.apache.mina.common.IoFilter.WriteRequest;
import org.apache.mina.common.support.BaseIoSession;
import org.apache.mina.common.support.DefaultWriteFuture;

import com.cqcis.uip.base.framework.FrameworkConstants;
import com.cqcis.uip.base.framework.UIPServiceException;
import com.cqcis.uip.base.framework.UIPTaskException;
import com.cqcis.uip.base.framework.codec.DataFormatException;
import com.cqcis.uip.base.framework.common.ControlMsg;
import com.cqcis.uip.base.framework.common.InterfaceLog;
import com.cqcis.uip.base.framework.common.State;
import com.cqcis.uip.base.framework.common.Task;
import com.cqcis.uip.base.framework.core.Endpoint;
import com.cqcis.uip.base.framework.core.RegisteredService;
import com.cqcis.uip.base.framework.core.RegisteredServiceContext;
import com.cqcis.uip.base.framework.core.SessionDispatcher;
import com.cqcis.uip.base.framework.core.SocketClientTaskContextAfterProcessor;
import com.cqcis.uip.base.framework.core.impl.AbstractTaskContext;
import com.cqcis.uip.base.framework.utils.ConcurrentTaskQueue;
import com.cqcis.uip.base.framework.utils.LogServiceThreadPoolExecutor;
import com.cqcis.uip.base.framework.utils.LoopThreadPoolExecutor;
import com.cqcis.uip.base.framework.utils.SessionUtil;
import com.cqcis.uip.base.framework.utils.StringUtil;
import com.cqcis.uip.base.framework.utils.TaskUtil;
import com.cqcis.uip.base.framework.utils.TimeUtil;
import com.cqcis.uip.base.framework.utils.LogServiceThreadPoolExecutor.LogThreadPoolExecutor;
import com.cqcis.uip.base.logadaptor.LogConfiguration;

/**
 * @author huangmy
 * @date   2007-12-14
 */
public abstract class AbstractClientHandler extends AbstractSocketHandler implements SocketClientTaskContextAfterProcessor {
	private static Log log = LogFactory.getLog(AbstractClientHandler.class);
	private static String READ_TIMEOUT_BEGIN = "READ_TIMEOUT_BEGIN" + AbstractClientHandler.class.getName();
	private boolean keepAlive;      // 是否长连接
	private LoopThreadPoolExecutor ltpeForTasksProcess;// 获取task进行发送到session的线程,避免handler线程池阻塞
	private LoopThreadPoolExecutor ltpeForSessionsProcess; // for bind session
	private LogThreadPoolExecutor timeoutCheckExecutor; // 超时检测
	private Object timeoutCheckLock = new Object();
	private Set<IoSession> readTimeoutCheckSessions = Collections.synchronizedSet(new HashSet<IoSession>());
	private ConcurrentTaskQueue<IoSession> waitForTaskSessions;
	private ConcurrentTaskQueue<IoSession> waitForBindSessions; // for bind session
	private int threadPoolCoreSize = 3;
	private int threadPoolMaxSize = 16;
	private int threadPoolKeepAliveTime = 60 * 1000;
	private RegisteredServiceContext serviceContext; // 上下文
    private final static String TASK_FINISHED_WAIT = AbstractClientHandler.class.getName() + "_TASK_FINISHED_WAIT";
    private final static String BLOCKING_FOR_GETTASK = AbstractClientHandler.class.getName() + "_BLOCKING_FOR_GETTASK";
    private final static String SESSION_PROCESS_LOCK = AbstractClientHandler.class.getName() + "_SESSION_PROCESS_LOCK";
	private boolean closeSessionIfTimeout;		// 超时后是否关闭链接
	private SessionDispatcher sessionDispatcher; // 分派链接
	    
	public void registerPoolProps(int threadPoolCoreSize, int threadPoolMaxSize, int threadPoolKeepAliveTime) {
		this.threadPoolCoreSize = threadPoolCoreSize;
		this.threadPoolMaxSize = threadPoolMaxSize;
		this.threadPoolKeepAliveTime = threadPoolKeepAliveTime;
	}
	
	@Override
	public void start(RegisteredServiceContext serviceContext) throws Exception {		
		super.start(serviceContext);
		
		this.ltpeForTasksProcess = LoopThreadPoolExecutor.getExecutor(threadPoolCoreSize, threadPoolMaxSize, threadPoolKeepAliveTime, serviceContext.getLogService(), 
				getLogConfiguration(serviceContext), "Dispatch Not Bind Sessions " + getLogThreadSuffixName(serviceContext), threadPoolKeepAliveTime);
		this.ltpeForSessionsProcess = LoopThreadPoolExecutor.getExecutor(threadPoolCoreSize, threadPoolMaxSize, threadPoolKeepAliveTime, serviceContext.getLogService(), 
				getLogConfiguration(serviceContext), "Dispatch Bind Sessions " + getLogThreadSuffixName(serviceContext), threadPoolKeepAliveTime);
		this.waitForTaskSessions = new ConcurrentTaskQueue<IoSession>();
		this.waitForBindSessions = new ConcurrentTaskQueue<IoSession>();
		this.serviceContext = serviceContext;
		this.timeoutCheckExecutor = (LogThreadPoolExecutor)LogServiceThreadPoolExecutor.getLogExecutor(1, 2, threadPoolKeepAliveTime, serviceContext, "ReadTimeoutCheck " + getLogThreadSuffixName(serviceContext));
	}
	
	protected LogConfiguration getLogConfiguration(RegisteredServiceContext serviceContext) {
		RegisteredService rs = serviceContext.getRegisteredService();
		if (rs != null) {
			return rs.getLogConfiguration();
		}
		return null;
	}
	
	protected String getLogThreadSuffixName(RegisteredServiceContext serviceContext) {
		RegisteredService rs = serviceContext.getRegisteredService();
		if (rs != null) {
			return rs.getName();
		}
		return "";
	}
	
	@Override
	public void stop(RegisteredServiceContext serviceContext) throws Exception {
		if (this.ltpeForTasksProcess != null) {
			this.ltpeForTasksProcess.shutdown();
		}
		
		if (this.ltpeForSessionsProcess != null) {
			this.ltpeForSessionsProcess.shutdown();
		}
		
		if (this.timeoutCheckExecutor != null) {
			this.timeoutCheckExecutor.shutdown();
		}
		
		super.stop(serviceContext);
	}
	
	/**
	 * 未绑定session获取task，然后处理
	 * 
	 */
	private void doGetAndProcessTaskForSessionsNotBind(ConcurrentTaskQueue<IoSession> sessions) {
		IoSession session = null;
		while((session = sessions.poll()) != null) {			
			if (session.isClosing()) {// 链接无效				
				continue;
			}
			
			/* continue to get task for the special session */
			Task taskNext = null;
			try {
				Task task = SessionUtil.getBindTask(session);
				if (task != null) { // bind task
					taskNext = processBindTaskForSession(session, task);
				} else {
			        taskNext = tryGetTaskForSession(session);
				}
			} catch (Throwable t) {
				SessionUtil.removeSessionStateWaitingTask(session);
				log.error("Catch Throwable in get task for session.", t);
				continue;
			}
			
			SessionUtil.removeSessionStateWaitingTask(session);
			
			if (taskNext != null) {				
				try {
					doProcessTaskForSession(session, taskNext);
				} catch (Throwable e) {
					session.setAttribute(SESSION_SERVICE_EXCEPTION, Boolean.TRUE);
					try {
						exceptionCaught(session, e);
					} catch (Throwable t) {
						log.error("Catch Throwable in dispatching task to sessions.", t);
					}
				}
			} else if (!session.isClosing() && SessionUtil.getBindTask(session) == null) {
				idleSessionTo(session);
			}
		}
	}
	
	@Override
	protected void doExceptionCaught(IoSession session, Throwable cause)
			throws Exception {
		removeFromCheckReadTimeout(session);
		Task task = SessionUtil.removeCurrentTask(session);
		SessionUtil.removeBindTask(session);
		
		String errMsg = null;
		if (task != null) {			
			TaskUtil.setTaskStateInValid(task);
			AbstractTaskContext<?> taskContext = (AbstractTaskContext<?>)task.getTaskContext();
			if (taskContext != null) {
				taskContext.interrupt();
			}
			
			TaskUtil.notifyTaskWaiter(task);
			
			if (TaskUtil.isFinishedAgainTask(task)) {
				dispatchFinishedTask(task);
				notifyFinishWaitIfMaybe(task);
				return;
			}
			
			String message = null;
			if (cause.getCause() != null && cause.getCause().getCause() instanceof DataFormatException) {
				message = FrameworkConstants.TASK_STATE_MSG_REMOTE_DATA_FORMAT_ERROR + ":" + cause.getMessage();
			} else {
				String preMsg = (cause instanceof UIPTaskException) ? "" : (FrameworkConstants.TASK_STATE_MSG_INTERFACE_PROCESS_ERROR + ":");
				message = preMsg + cause.getMessage();			    
			}
			
			task.setThrowable(cause);
												
			dispatchTaskExceptionResult(task, message, cause); // 异常结果
			notifyFinishWaitIfMaybe(task);
			
			errMsg = message;
			
			log.error("Exception Task: " + task);
		}
		
		try {						
			InterfaceLog intflog = getIntfLogAndSetRecvRecord(session, true);		
			if (intflog != null) {
				String recvRecord = intflog.getRecvRecord();
				if (!StringUtil.hasText(recvRecord)) {
					intflog.setRecvRecord(errMsg);
				}
				getInterfaceLogService().insert(intflog);
			}
		} catch (Exception e) {
			log.error("Save tin_log Exception:", e);
		}
	}
	
	/**
	 * 二进制日志路径
	 * 
	 * @param file
	 * @return
	 */
	protected String getRecordPathFrom(String file) {
		if (file == null) {
			return null;
		}
		int splitBegin = file.indexOf(FrameworkConstants.STREAM_LOG_PATH_STOKEN);
		if (splitBegin >= 0) {
			splitBegin = splitBegin + FrameworkConstants.STREAM_LOG_PATH_STOKEN.length() + 1;
		} else {
			splitBegin = 0;
		}
		
		return file.substring(splitBegin);
	}
	
	/**
	 * 获取接口日志对象并且设置接收路径
	 * 
	 * @param session
	 * @return
	 */
	private InterfaceLog getIntfLogAndSetRecvRecord(IoSession session, boolean remove) {
		if (session.getAttribute(FrameworkConstants.INTERFACE_STREAM_RECORD_CLOSED) != null) {
			return null;// 关闭字节流记录功能
		}
		
		String file = (String)session.getAttribute(FrameworkConstants.INTERFACE_STREAM_RECORD);	
		InterfaceLog intflog = remove ? (InterfaceLog)session.removeAttribute(FrameworkConstants.INTERFACE_LOG)
				: (InterfaceLog)session.getAttribute(FrameworkConstants.INTERFACE_LOG);			
		if (intflog != null && file != null) {
			intflog.setRecvTime(TimeUtil.getCurrentTime());
			intflog.setRecvRecord(getRecordPathFrom(file));
		}
		
		return intflog;
	}
	
	/**
	 * 产生接口日志对象，并获取发送路径
	 * 
	 * @param session
	 * @param task
	 * @return
	 */
	private InterfaceLog createIntfLogAndSetSendRecord(IoSession session, Task task) {
		String sendTime = TimeUtil.getCurrentTime();				
		String intfSeq = (String)task.getAttribute(FrameworkConstants.LOG_RECORDID);
		
		intfSeq = intfSeq == null ? task.getTaskId() : intfSeq;
		task.setAttribute(FrameworkConstants.LOG_RECORDID, intfSeq);
		
	    InterfaceLog intflog = new InterfaceLog(intfSeq, getInterfaceId());
	    session.setAttribute(FrameworkConstants.INTERFACE_LOG, intflog);
	    if (session.getAttribute(FrameworkConstants.INTERFACE_STREAM_RECORD_CLOSED) != null) {
			return null;// 关闭字节流记录功能
		}
	    	    
	    intflog.setSendTime(sendTime);			  
	    String file = (String)session.removeAttribute(FrameworkConstants.INTERFACE_STREAM_RECORD);
	    intflog.setSendRecord(getRecordPathFrom(file));	
	    
		return intflog;
	}
	
	@Override
	protected void doMessageReceived(IoSession session, Object message) throws Exception {
		removeFromCheckReadTimeout(session);
		synchronized (session.getAttribute(SESSION_PROCESS_LOCK)) {
			forMessageReceived(session, message);
		}
	}
		
	/**
	 * 处理接受数据
	 * 
	 * @param session
	 * @param message
	 * @throws Exception
	 */
	private void forMessageReceived(IoSession session, Object message) throws Exception {		
		boolean isControlMsg = false;
		Task task = SessionUtil.getCurrentTask(session);
		if (task == null) {			
			if (message instanceof ControlMsg) {
				isControlMsg = true;
			}
		}
		
		try {		
			if (!isControlMsg || (isControlMsg && ((ControlMsg)message).getMessage() != null)) {
				InterfaceLog intflog = getIntfLogAndSetRecvRecord(session, false);			
				if (intflog != null) {				
					getInterfaceLogService().insert(intflog);
				}
			}			
		} catch (Exception e) {
			log.error("Save tin_log Exception:", e);
		}
		
		if (task == null) {
			session.close();
			if (log.isDebugEnabled()) {
				log.debug("当前链接上没有对应的task，链接视为接收到无效数据，关闭.MESSAGE: " + message);
			}
			return; // 不处理
		}
		
		boolean beforeGenerateResult = false;
		int beforeGenerateResultCount = 0;
		int oldStateHisSize = 0;
		int beforeAgainCount = 0;

		beforeGenerateResult = task.isGenerateResult();
		beforeGenerateResultCount = task.getGenerateResultCount();
		oldStateHisSize = task.getStateHistory().size();
		beforeAgainCount = task.getAgainCount();
		task.setExecuteDefault(true);

        // 服务
		doMessageReceived(message, task); // 调用自定义服务
		
		// 后续处理
		forReleaseAfterMsgRecved(task, beforeGenerateResult, beforeGenerateResultCount, 
				oldStateHisSize, beforeAgainCount, session, false);
	}
	
	public void afterNormalFinished(Task task, IoSession session) {
		synchronized (session.getAttribute(SESSION_PROCESS_LOCK)) {
		    forReleaseAfterMsgRecved(task, task.isGenerateResult(), task.getGenerateResultCount(), 
			    	task.getStateHistory().size(), task.getAgainCount(), session, true);
		}
	}
	
	/**
	 * 后续处理
	 * 
	 * @param task
	 * @param beforeGenerateResult
	 * @param beforeGenerateResultCount
	 * @param oldStateHisSize
	 * @param beforeAgainCount
	 * @param session
	 */
	private void forReleaseAfterMsgRecved(Task task, boolean beforeGenerateResult, int beforeGenerateResultCount, 
			int oldStateHisSize, int beforeAgainCount, IoSession session, boolean after) {
		// 是否生成结果
		boolean generateResult = TaskUtil.isGenerateResultForCompareWithBefore(task, 
				beforeGenerateResultCount, beforeGenerateResult);		
		boolean agained= TaskUtil.hasSetAgainFor(task, beforeAgainCount); // AGAIN	
		
		if (TaskUtil.isFinishedAgainTask(task)) {
			dispatchFinishedTask(task);
			if (generateResult || agained) {
				throw new UIPServiceException("已经处理结束的task不允许再设置again或者generateResult为true.");
			}
		}
		
		if (log.isDebugEnabled()) {
			log.debug("doMessageReceived: agained = " + agained  + ", exclusive = " + task.isExclusive(agained) 
					+ ", generateResult = " + generateResult
					+ ", executeDefault = " + task.isExecuteDefault()
					+ (after ? ", AfterNormalFinished" : ""));
		}
		
		if (generateResult) {// 产生结果
			if (task.getStateHistory().size() == oldStateHisSize) { // state not been changed
				task.setState(State.SUCCESS);
				task.setMessage(FrameworkConstants.TASK_STATE_MSG_SUCESS);
			}
		}
				
		if (agained && generateResult) {// AGAIN, RESULT -> 先again，最终返回结果			
			TaskUtil.setFinishedAgainTaskFlag(task);			
			if (getTaskFinishWaitLock(task) == null) {// 如果没有设置again，则设置again				
				doAgainForTask(task, session, agained);
			}
			notifyFinishWaitIfMaybe(task); // task处理完成
			return;
		} else if (agained && !generateResult) {// AGAIN, NOT RESULT -> again
			doAgainForTask(task, session, agained);
			notifyFinishWaitIfMaybe(task); // task处理完成
			return;
		} else if (!agained && generateResult) {// NOT AGAIN, RESULT -> result			
			dispatchFinishedTask(task);
		} else if (!agained && !generateResult) {// no
			if (task.isExecuteDefault()) {
				SessionUtil.removeBindTask(session); // 解除相关联的task
				TaskUtil.removeBindSession(task); // 解除相关联的session	
				idleSessionTo(session);
			}
						
			notifyFinishWaitIfMaybe(task); // task处理完成									
			return;
		}
		
		// 生成结果
		releaseSessionAndNotifyTask(session, task); 
		// 不是长连接, 客户端接收数据后关闭session
		if (!this.keepAlive) {
			session.close();			
		}
	}
	
	/**
	 * 释放链接， 同时通知task处理完成
	 * 
	 * @param session
	 * @param task
	 */
	private void releaseSessionAndNotifyTask(IoSession session, Task task) {
		// 生成结果
        SessionUtil.removeCurrentTask(session);
		
		SessionUtil.removeBindTask(session); // 解除相关联的task
		TaskUtil.removeBindSession(task); // 解除相关联的session
		idleSessionTo(session);
		
		notifyFinishWaitIfMaybe(task); // task处理完成
	}
	
	/**
	 * 完成
	 * 
	 * @param task
	 */
	private void dispatchFinishedTask(Task task) {
		synchronized (task) {
			if (TaskUtil.isTaskHasGenerateFinished(task)) {
				return;
			}
			
			Task t = task.copy();// new
			t.setAttachement(task.getAttachement());
			
			Endpoint from = task.route().from(); // 来源
			Endpoint to = task.route().to(); // 目的
			// 反转
			t.route().to(from);
			t.route().from(to);
						
			if (TaskUtil.isSyncTask(task)) {
				TaskUtil.createOrGetSyncCTQFor(task,
						FrameworkConstants.TASK_SYNCCTQ_FOR_TASK).offer(t); // 同步
			} else {
				getSendTaskQueue().offer(t);
			}
			
			TaskUtil.setTaskHasGenerateFinished(task);
		}
	}
	
	/**
	 * 等待任务的连接
	 * 
	 * @param session
	 */
	protected void idleSessionTo(IoSession session) {
		removeFromCheckReadTimeout(session);
		if (session.isClosing()) {
			return;
		}

		if (SessionUtil.isSessionStateWaitingTask(session)) {
			return;
		}

		session.removeAttribute(FrameworkConstants.INTERFACE_LOG);
		SessionUtil.setSessionStateWaitingTask(session);// 等待
		session.setIdleTime(IdleStatus.READER_IDLE, 0);
		if (SessionUtil.getBindTask(session) != null) {
			idleBindSessionForTask(session);// bind
		} else {
			idleNotBindSessionForTask(session);// unbind
		}
	}
	
	/**
	 * 将绑定的空闲session入队等待，等待处理绑定的task
	 * 
	 * @param session
	 */
	private void idleBindSessionForTask(IoSession session) {
		idleSessionForTask(session, this.waitForBindSessions, this.ltpeForSessionsProcess);
	}
	
	/**
	 * 将未绑定的空闲session入队等待，获取task
	 * 
	 * @param session
	 */
	private void idleNotBindSessionForTask(IoSession session) {
		idleSessionForTask(session, this.waitForTaskSessions, this.ltpeForTasksProcess);
	}
	
	/**
	 * 将未绑定的空闲session在指定的队列上入队等待，获取task
	 * 
	 * @param session
	 * @param sessions
	 * @param workerExecutor
	 */
	private void idleSessionForTask(IoSession session, final ConcurrentTaskQueue<IoSession> sessions, LoopThreadPoolExecutor workerExecutor) {		
		sessions.offer(session);// 链接空闲
		// 判定是否新建工作线程
		if (workerExecutor.getPoolSize() <= 0 || sessions.size() > workerExecutor.getPoolSize() * 2) {
			workerExecutor.execute(new Runnable() {
				public void run() {
					doGetAndProcessTaskForSessionsNotBind(sessions);
				}
			});// 发'分派任务到连接'命令
		}
		
		sessions.notifyQueue();
		workerExecutor.notifyPool();	
	}
	
	@Override
	protected void doMessageSent(IoSession session, Object message) throws Exception {
		if (FrameworkConstants.VIRTUAL_TRY_TIMEOUT_BUFFERS.remove(message)) {
			return;
		}
		
		Task task = SessionUtil.getCurrentTask(session);
		try {
			if (timeout > 0 && !SessionUtil.isTimeoutCheckClosed(session)) {				
				((BaseIoSession)session).increaseIdleCount(IdleStatus.READER_IDLE);
				session.setIdleTime(IdleStatus.READER_IDLE, (int)(timeout/1000));
			}
			
			if (task != null) {
				createIntfLogAndSetSendRecord(session, task);
				
				if (!SessionUtil.isTimeoutCheckClosed(session)) {
				    addToforCheckReadTimeout(session);
				}
		    }		    
		} catch (Exception e) {
			log.error("Set tin_log Exception:", e);
		}
	}
	
	private WriteRequest getTryStartSocketIoProcessorRequest(IoSession session, ByteBuffer buf) {
		WriteFuture future = new DefaultWriteFuture(session);
		FrameworkConstants.VIRTUAL_TRY_TIMEOUT_BUFFERS.add(buf);
        return (new WriteRequest(buf, future, null));
	}
	
	private void addToforCheckReadTimeout(IoSession session) {
		session.setAttribute(READ_TIMEOUT_BEGIN, new Long(System.currentTimeMillis()));
		
		this.readTimeoutCheckSessions.add(session);
		if (this.readTimeoutCheckSessions.size() <= 1 || this.timeoutCheckExecutor.getPoolSize() <= 0) {				
			this.timeoutCheckExecutor.execute(new Runnable() {
				Object lock = new Object();
				public void run() {
					while (true) {
						synchronized (lock) {
							try {
								lock.wait(2 * 1000);
							} catch (InterruptedException e) {
								break;
							}
						}
						
						synchronized (timeoutCheckLock) {
							if (!runForCheckReadTimeout()) {
								break;
							}
						}						
					}
				}
			});
		}
		
			
	}
	
	private boolean removeFromCheckReadTimeout(IoSession session) {
		Object o = session.removeAttribute(READ_TIMEOUT_BEGIN);
		if (o != null) {
			this.readTimeoutCheckSessions.remove(session);
			return true;
		}
		return false;
	}
	
	/**
	 * 检查超时
	 * 
	 * @return true 继续运行
	 */
	private boolean runForCheckReadTimeout() {
		if (this.readTimeoutCheckSessions.isEmpty()) {
			return false;
		}
		
		IoSession[] sessions = new IoSession[this.readTimeoutCheckSessions.size()];		
		sessions = this.readTimeoutCheckSessions.toArray(sessions);
				
		for (IoSession session : sessions) {
			synchronized (session.getAttribute(SESSION_PROCESS_LOCK)) {
				checkReadTimeout(session);
			}
		}
		
		return true;
	}
	
	private void checkReadTimeout(IoSession session) {
		long currentTime = System.currentTimeMillis();
		Long readTimeoutBeginLong = (Long) session
				.getAttribute(READ_TIMEOUT_BEGIN);
		if (readTimeoutBeginLong != null) {
			long readTimeoutBegin = readTimeoutBeginLong;
			
			long timeInterval = currentTime - readTimeoutBegin;
			if (timeInterval > timeout && timeInterval <= (timeout + timeout / 5)) {
				WriteRequest wq = getTryStartSocketIoProcessorRequest(session, ByteBuffer.allocate(4).flip());
				session.getFilterChain().fireFilterWrite(session, wq);
			}
						
			if (timeInterval >= (timeout + timeout / 2)) {// timeout
				try {
					if (session.getAttribute(READ_TIMEOUT_BEGIN) != null) {
						forDoSessionIdle(session, IdleStatus.READER_IDLE, true);
						this.readTimeoutCheckSessions.remove(session);
					}

				} catch (Exception e) {
				}
			}
		}
	}

	@Override
	protected void doSessionCreated(IoSession session) throws Exception {
		if (!isStreamRecord()) {// 关闭字节流功能
			session.setAttribute(FrameworkConstants.INTERFACE_STREAM_RECORD_CLOSED, "true");
		}
		
		session.setAttribute(SESSION_PROCESS_LOCK, new Object());
		session.setAttribute(FrameworkConstants.SOCKET_CLIENT_TASK_AFTERPROCESSOR, this);
		
		if (this.sessionDispatcher != null && !sessionDispatcher.isToDispatch(session)) {
			return;
		}
		
		idleSessionTo(session);	// 发任务分派命令
	}
		
	/**
	 * 阻塞获取处理任务, 如果没有可用的任务, 则线程阻塞
	 * 
	 * @param session
	 * @return
	 */
	private Task tryGetTaskForSession(IoSession session) {
		return sessionGetTask(session, true);
	}
	
	/**
	 * 等待task处理完成
	 * 
	 * @param task
	 */
	private void waitTaskFinished(Task task) {
		CountDownLatch latch = (CountDownLatch)task.getAttribute(TASK_FINISHED_WAIT);		
		if (latch == null) {
			return;
		}
		
		try {
			latch.await();
		} catch (InterruptedException e) {			
		} finally {
			task.removeAttribute(TASK_FINISHED_WAIT);
		}
	}
	
	/**
	 * 设置task等待完成锁
	 * 
	 * @param task
	 */
	private boolean putTaskFinishWaitLock(Task task) {
		CountDownLatch latch = (CountDownLatch)task.getAttribute(TASK_FINISHED_WAIT);
		if (latch != null) {
			return false;// 已经设置，不允许
		}
		latch = new CountDownLatch(1);
		task.setAttribute(TASK_FINISHED_WAIT, latch);
		return true;
	}
	
	/**
	 * 锁
	 * 
	 * @param task
	 * @return
	 */
	private CountDownLatch getTaskFinishWaitLock(Task task) {
		return (CountDownLatch)task.getAttribute(TASK_FINISHED_WAIT);
	}
	
	/**
	 * 可能的话，通知等待任务，处理完成
	 * 
	 * @param task
	 */
	private void notifyFinishWaitIfMaybe(Task task) {		
		CountDownLatch latch = (CountDownLatch)task.getAttribute(TASK_FINISHED_WAIT);
		if (latch != null) {
			latch.countDown();
		}
	}
	
	/**
	 * 处理绑定的task
	 * 
	 * @param session
	 * @return
	 */
	private Task processBindTaskForSession(IoSession session, Task task) {
		// 1. 等待前一task完成
		waitTaskFinished(task);
		
		// 2. 处理等待		
		if (!doOperationForTimeConsumeIfAgain(task)) {// wait
			// 放入了等待池处理				
			return null; // 认为没有找到task
		}	
		
		return task;
	}
	
	/**
	 * 非阻塞获取处理任务, 如果没有可用的任务，当前线程立即返回
	 * 
	 * @param session
	 * @return
	 */
	protected Task tryGetTaskForSessionNoBlocking(IoSession session) {				
		return sessionGetTask(session, false);
	}
	
	/**
	 * 获取任务
	 * 
	 * @param session
	 * @param block
	 * @return
	 */
	private Task sessionGetTask(IoSession session, boolean block) {		
		Task task  = null;
		try {
			session.setAttribute(BLOCKING_FOR_GETTASK, Thread.currentThread());
			task = block ? getTask() : getTaskNoBlocking();
		} finally {
			session.removeAttribute(BLOCKING_FOR_GETTASK);
		}
		
		if (task == null) {			
			return null;
		}
		
		TaskUtil.setTimeout(task, getTimeout());// 超时时间
		
		// 1. 等待前一task完成
		waitTaskFinished(task);
		
		// 2. 处理等待
		try {
			TaskUtil.removeCurrentSession(task);

			if (!doOperationForTimeConsumeIfAgain(task)) {// wait
				// 放入了等待池处理				
				return null;// 认为没有找到task
			}

			if (session.isClosing()) { // session关闭
				if (maybeReProcessFor(task, session)) {
					return null;// null对应的session表示没有取到task
				}
			}
		} catch (Throwable t) {
			dispatchTaskExceptionResult(task, t.getMessage(), t); // 异常结果
			return null;
		}

		if (TaskUtil.isTaskStateInValid(task)) {// 失效
			SessionUtil.removeBindTask(session); // 解除相关联的task
			TaskUtil.removeBindSession(task); // 解除相关联的session
			return null;
		}

		TaskUtil.setCurrentSession(task, session); // set session
		TaskUtil.removePendingWait(task);
				
		return task;
	}
	
	/**
	 * 可能重处理,
	 * 
	 * @param task
	 * @param session
	 * @return true 如果重排入队列处理.
	 */
	private boolean maybeReProcessFor(Task task, IoSession session) {
		Set<IoSession> sessions = getAllSessions(session);
		if (sessions != null) {
			if (sessions.size() > 0) {
				getTaskQueue().offer(task);// 重新排入
				return true;
			}

			// 未决提交
			boolean hasPendingSchedule = this.serviceContext
					.getRegisteredService().hasPendingSchedule();
			if (hasPendingSchedule && !TaskUtil.isSyncTask(task)) {
				getTaskQueue().offer(task);// 重新排入
				return true;
			}
		}
		
		return false;
	}
	
	/**
	 * 重处理Task
	 * 
	 * @param task
	 * @param session
	 * @return
	 */
	private boolean doAgainForTask(Task task, IoSession session, boolean agained) {	
		if (!agained) {
			return false;
		}
        		
		if (TaskUtil.isPendingWait(task)) {
			return true;// 已经是未决，直接返回
		}
		
		boolean b = putTaskFinishWaitLock(task);
		if (!b) {// 已经设置
			throw new UIPServiceException("task已经被设置Again，在该操作完成之前不允许继续设置");
		}
		
		TaskUtil.setPendingWait(task);// 未决等待
		if (task.isExclusive(agained)) {					
			SessionUtil.setBindTask(session, task); // session 绑定 task
			TaskUtil.setBindSession(task, session);// 同时task 也绑定 session
		} else {		
			SessionUtil.removeCurrentTask(session);
			TaskUtil.removeCurrentSession(task);
			SessionUtil.removeBindTask(session); // 解除相关联的task
			TaskUtil.removeBindSession(task); // 解除相关联的session
		    getTaskQueue().offer(task);				    
		}
				
		idleSessionTo(session);
		
		return true;
	}
	
	/**
	 * 检查session和task是否正常
	 * 
	 * @param session
	 * @param task
	 * @return
	 */
	private boolean checkValidForSessionAndTask(IoSession session, Task task) {
		if (TaskUtil.isTaskStateInValid(task)) {//失效
			SessionUtil.removeCurrentTask(session);
			SessionUtil.removeBindTask(session); // 解除相关联的task
			TaskUtil.removeBindSession(task); // 解除相关联的session
			idleSessionTo(session);
			return false;
		}
		
		if (session.isClosing()) {	
			if (maybeReProcessFor(task, session)) {
				return false;// 如果重处理了, 返回false
			}
			
			String message = SessionUtil.isClosedByService(session) 
	            ? FrameworkConstants.TASK_STATE_MSG_SERVICE_FORCE_CLOSED_SESSION
			    : FrameworkConstants.TASK_STATE_MSG_REMOTE_CLOSS_SESSION;
		    
			TaskUtil.notifyTaskWaiter(task);
			notifyFinishWaitIfMaybe(task);
			dispatchTaskExceptionResult(task, message, null);
			return false;
		}
		
		return true;
	}
	
	/**
	 * 处理
	 * 
	 * @param session
	 * @param task
	 * @throws Exception
	 */
	public void doProcessTaskForSession(IoSession session, Task task) throws Exception {
		synchronized (session.getAttribute(SESSION_PROCESS_LOCK)) {
			try {
			    forProcessTaskForSession(session, task);
			} finally {
				CountDownLatch countLatch = (CountDownLatch)task.removeAttribute(FrameworkConstants.TASK_BLOCKINNG_PROCESS_SYNC);
				if (countLatch != null) {
					countLatch.countDown();
				}
			}
		}
	}
	
	/**
	 * 处理
	 * 
	 * @param session
	 * @param task
	 * @throws Exception
	 */
	private void forProcessTaskForSession(IoSession session, Task task) throws Exception {			
		if (!checkValidForSessionAndTask(session, task)) {
			return;
		}
				
		SessionUtil.setCurrentTask(session, task);// 当前task
		
		boolean beforeGenerateResult = false;
		int beforeGenerateResultCount = 0;				
		int beforeAgainCount = 0;
		Object obj = null;

		beforeGenerateResult = task.isGenerateResult();
		beforeGenerateResultCount = task.getGenerateResultCount();
		task.setExecuteDefault(true); // 设置为执行默认的系统行为

		beforeAgainCount = task.getAgainCount();
		obj = doMessageSend(task); // 调用定义服务
		
		boolean agained = TaskUtil.hasSetAgainFor(task, beforeAgainCount); // AGAIN
		boolean generateResult = TaskUtil.isGenerateResultForCompareWithBefore(task, beforeGenerateResultCount, beforeGenerateResult);
		
		if (log.isDebugEnabled()) {
			log.debug("doMessageSend: agained = " + agained + ", exclusive = " + task.isExclusive(agained) + 
					", generateResult = " + generateResult
					+ ", executeDefault = " + task.isExecuteDefault()
					+ ", sendObj is " + ((obj != null) ? "not null" : "null"));
		}
		
		if (agained || generateResult) {
			if (TaskUtil.isFinishedAgainTask(task)) {
				dispatchFinishedTask(task);	
				throw new UIPServiceException("已经处理结束的task不允许再设置again或者generateResult为true.");
			}
		}
		
		Task nowTask = SessionUtil.getCurrentTask(session);
		if (nowTask == null) {
			return;// 为空，不处理
		}
				
		if (obj != null) {
			Task writeTask = getAfterTask(obj);			
			if (writeTask != null) {
				String recordId = (String)writeTask.getAttribute(FrameworkConstants.LOG_RECORDID);
			    task.setAttribute(FrameworkConstants.LOG_RECORDID, recordId);
			}
			
			doAgainForTask(task, session, agained); // 重处理Task
				
			// 发送
			session.write(obj);
			
			if (!generateResult) {				
				return;
			}
			
			dispatchTaskDealingResult(task);
		} else {
			if (TaskUtil.isFinishedAgainTask(task)) {
				dispatchFinishedTask(task);	
				if (log.isWarnEnabled()) {
					log.warn("业务层doMessageSend方法在task处理结束后，继续处理返回null，该返回值对应操作没有定义。");
				}
			}
			
			if (!doAgainForTask(task, session, agained)) { // 重处理Task	
				if (!task.isExecuteDefault()) {
					if (!generateResult) {	
						if (SessionUtil.getBindTask(session) == null
								&& task.getTaskContext() == null) {	
							idleSessionTo(session);
						}
						
						return;
					}
					dispatchTaskDealingResult(task);
					
					if (SessionUtil.getBindTask(session) == null 
							&& task.getTaskContext() == null) {	
						idleSessionTo(session);
					}
					
					return;
				}
				
				throw new UIPServiceException("the method of 'doMessageSend' or call stack method can't return null when the task not be set again.");
			}
		}
	}
	
	/**
	 * 正处理
	 * 
	 * @param task
	 */
	private void dispatchTaskDealingResult(Task task) {
		synchronized (task) {
			if (TaskUtil.isHasGenerateExceptionResult(task)) {
				return;
			}
			
			if (TaskUtil.isHasGenerateResult(task)) {
				return;
			}
			
			task.setMessage(FrameworkConstants.TASK_STATE_MSG_REMOTE_DEALNG);
			task.setState(State.SUCCESS);
			
			TaskUtil.removeHasGenerateExceptionResult(task);
			
			dispatchTaskResult(task, false); // 正常处理返回结果
			TaskUtil.setHasGenerateResult(task);
		}		
	}
	
	@Override
	protected void doSessionClosed(IoSession session) throws Exception {	
		removeFromCheckReadTimeout(session);
		Thread blockSessionThread = (Thread)session.getAttribute(BLOCKING_FOR_GETTASK);
		if (blockSessionThread != null) {
			blockSessionThread.interrupt();
		}
		
		if (session.getAttribute(AbstractTaskContext.DONT_REMOVETASK_SESSION) != null) {
			return;
		}
		
		Task task = SessionUtil.removeCurrentTask(session);
		SessionUtil.removeBindTask(session);
		String errMsg = null;
		
		if (task != null) {
			TaskUtil.setTaskStateInValid(task);
			AbstractTaskContext<?> taskContext = (AbstractTaskContext<?>)task.getTaskContext();
			if (taskContext != null) {
				taskContext.interrupt();
			}
			
			TaskUtil.notifyTaskWaiter(task);
			
			if (TaskUtil.isFinishedAgainTask(task)) {
				dispatchFinishedTask(task);
				notifyFinishWaitIfMaybe(task);
				return;
			}
			
			String message = SessionUtil.isClosedByService(session) 
			        ? FrameworkConstants.TASK_STATE_MSG_SERVICE_FORCE_CLOSED_SESSION
					: FrameworkConstants.TASK_STATE_MSG_REMOTE_CLOSS_SESSION;
			
			dispatchTaskExceptionResult(task, message, null); // 异常结果
			notifyFinishWaitIfMaybe(task);
			
			errMsg = message;
			
			if (log.isDebugEnabled()) {
				log.debug("Session Closed for task: " + task);
			}
		}
		
		try {						
			InterfaceLog intflog = getIntfLogAndSetRecvRecord(session, true);	
			if (intflog != null) {		
				String recvRecord = intflog.getRecvRecord();
				if (!StringUtil.hasText(recvRecord)) {
					intflog.setRecvRecord(errMsg);
				}
				getInterfaceLogService().insert(intflog);
			}		
		} catch (Exception e) {
			log.error("Save tin_log Exception:", e);
		}
	}
	
	@Override
	protected void doSessionIdle(IoSession session, IdleStatus status)
			throws Exception {
		synchronized (session.getAttribute(SESSION_PROCESS_LOCK)) {
			forDoSessionIdle(session, status, false);
		}
	}
	
	private void forDoSessionIdle(IoSession session, IdleStatus status, boolean closeMust)
			throws Exception {		
		removeFromCheckReadTimeout(session);
		Task task = SessionUtil.getCurrentTask(session);
		if (task != null && status == IdleStatus.READER_IDLE) {
			TaskUtil.setTaskStateInValid(task);
			AbstractTaskContext<?> taskContext = (AbstractTaskContext<?>)task.getTaskContext();
			if (taskContext != null) {
				taskContext.interrupt();
			}
			SessionUtil.removeCurrentTask(session);	
			SessionUtil.removeBindTask(session);
			
			TaskUtil.notifyTaskWaiter(task);
			
			if (TaskUtil.isFinishedAgainTask(task)) {
				dispatchFinishedTask(task);
				notifyFinishWaitIfMaybe(task);
				return;
			}
			
			String message = FrameworkConstants.TASK_STATE_MSG_INTERFACE_TIMEOUT;
			message = message + (closeMust ? "-UIP" : "-SOCKET");
		    		   
		    dispatchTaskExceptionResult(task, message, null); // 异常结果
		    notifyFinishWaitIfMaybe(task);
		    
		    if (this.closeSessionIfTimeout || closeMust) {
		       session.close();
		    } else {
		    	idleSessionTo(session);
		    }
		    if (log.isDebugEnabled()) {
				log.debug("Session Idle for task: " + task
						+ ", IdleCheck: " + (closeMust ? "UIP" : "SOCKET"));
			}
		    
		    try {						
				InterfaceLog intflog = getIntfLogAndSetRecvRecord(session, true);	
				if (intflog != null) {		
					String recvRecord = intflog.getRecvRecord();
					if (!StringUtil.hasText(recvRecord)) {
						intflog.setRecvRecord(message);
					}
					getInterfaceLogService().insert(intflog);
				}		
			} catch (Exception e) {
				log.error("Save tin_log Exception:", e);
			}
		}
	}
	
	/**
	 * Session 收到数据后, 对应的数据处理操作;
	 * 返回任务处理结果;	 
	 * @param packetInfo
	 * @return
	 * @throws Exception
	 */
	public abstract void doMessageReceived(Object message, Task task) throws Exception;
	
	/**
	 * 处理给定的orderId对应的发送数据, 打包成特定接口对应的数据实例;
	 * 如果处理成功, 返回对应的实例; 如果失败, 必须返回null,表示该工单处理失败 
	 * @param orderId
	 * @return
	 * @throws Exception
	 */
	public abstract Object doMessageSend(Task task) throws Exception;
	
	public void registerKeepAlive(boolean keepAlive) {
		this.keepAlive = keepAlive;
	}

	public void registerCloseSessionIfTimeout(boolean closeSessionIfTimeout) {
		this.closeSessionIfTimeout = closeSessionIfTimeout;
	}

	public void registerSessionDispatcher(SessionDispatcher sessionDispatcher) {
		this.sessionDispatcher = sessionDispatcher;
	}
}